package com.sojson.util.mq.rabbit.code.interfaces;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.GetResponse;
import com.sojson.util.mq.rabbit.interfaces.IRabbitBaseConsumeUtil;

/**
 * Rabbit消费消息工具类接口
 * 
 * @author liu
 * @date 2022-06-16
 */
public interface IRabbitCodeConsumeUtil extends IRabbitBaseConsumeUtil {

    /**
     * 开始消费消息
     * 
     * @param channel
     * @param queue     队列名
     * @param callback  消息的回调接口,创建Consumer接口的实现类DefaultConsumer,重写handleDelivery方法,去处理队列中的数据
     * @return
     * @throws IOException
     */
    default String basicConsume(Channel channel, String queue, Consumer callback) throws IOException {
        return basicConsume(channel, queue, false, queue, callback);
    }

    /**
     * 开始消费消息
     * 
     * @param channel
     * @param queue     队列名
     * @param autoAck   是否开启自动确认机制</BR>
     *                  当rabbitmq将消息发送给消费者进行消费时,消费者可能会消费消息失败的情况,用户可以设置消费失败的消息给其他消费者消费或者直接丢弃</br>
     *                  自动确认模式积极的一面是能够拥有更高的吞吐量,但是却存在数据安全的问题</br>
     * @param callback  消息的回调接口,创建Consumer接口的实现类DefaultConsumer,重写handleDelivery方法,去处理队列中的数据
     * @return
     * @throws IOException
     */
    default String basicConsume(Channel channel, String queue, boolean autoAck, Consumer callback) throws IOException {
        return basicConsume(channel, queue, autoAck, queue, callback);
    }

    /**
     * 开始消费消息
     * 
     * @param channel
     * @param queue         队列名
     * @param autoAck       是否开启自动确认机制</BR>
     *                      当rabbitmq将消息发送给消费者进行消费时,消费者可能会消费消息失败的情况,用户可以设置消费失败的消息给其他消费者消费或者直接丢弃</br>
     *                      自动确认模式积极的一面是能够拥有更高的吞吐量,但是却存在数据安全的问题</br>
     * @param consumerTag   收到消息时会传到callback的handleDelivery()
     * @param callback      消息的回调接口,创建Consumer接口的实现类DefaultConsumer,重写handleDelivery方法,去处理队列中的数据
     * @return
     * @throws IOException
     */
    default String basicConsume(Channel channel, String queue, boolean autoAck, String consumerTag, Consumer callback)
        throws IOException {
        // 该模式的主体是队列,采用的是发布订阅的模式
        // 当队列中有新的消息时,队列就会主动将消息推送给消费者进行消费
        // 它保证了消费者能够及时的获取到相关的消息,其建立道德连接是长连接,比较节约资源,官方推荐使用该方法
        return channel.basicConsume(getQueueName(queue), false, consumerTag, callback);
    }

    /**
     * 获取消息内容
     * 
     * @param channel
     * @param queue     队列名
     * @return
     * @throws IOException
     */
    default String getBody(Channel channel, String queue) throws IOException {
        return getBody(channel, queue, true);
    }

    /**
     * 获取消息内容
     * 
     * @param channel
     * @param queue     队列名
     * @param autoAck   是否开启自动确认机制</BR>
     *                  当rabbitmq将消息发送给消费者进行消费时,消费者可能会消费消息失败的情况,用户可以设置消费失败的消息给其他消费者消费或者直接丢弃</br>
     *                  自动确认模式积极的一面是能够拥有更高的吞吐量,但是却存在数据安全的问题</br>
     * @return
     * @throws IOException
     */
    default String getBody(Channel channel, String queue, boolean autoAck) throws IOException {
        // 该模式的主体是消费者,当消费者需要队列中的一条消息时,可以通过该方法去获取到队列中的一条消息
        // 其建立的连接是短连接,当消费者获取到消息后该连接断开
        // 可以通过循环的方式实现和basicConsume相同的效果,但不建议这样去使用
        GetResponse work = channel.basicGet(queue, autoAck);
        return new String(work.getBody());
    }

    /**
     * 初始化基础消费者
     * 
     * @param channel
     * @throws IOException
     */
    default void initBase(Channel channel) throws IOException {
        /**
         * 当前消费者一次消费多少条消息(先取到一定数量,然后重复调用当前消费者)</BR>
         * 设置通道的预取数量为1，官方推荐100到300，数据会影响其吞吐量</BR>
         * 
         * @param prefetchSize  可接收消息的大小的
         * @param prefetchCount 客户端从服务端每次取消息的数量,如果这些消息没有确认或拒绝,服务端就不会发消息给这个客户端</BR>
         * @param global        一个Connection可以有多个Channel,如果是false则说明只是针对于这个Channel的
         * @throws IOException
         */
        channel.basicQos(0, PREFETCH_COTCHUNT, true);
    }

    /**
     * 初始化消费者
     * 
     * @param channel
     * @throws IOException
     */
    default void init(Channel channel) throws IOException {
        initBase(channel);
    }

}